// This file is part of OpenTSDB.
// Copyright (C) 2017 TeleDB Development Group.
//
// This program is free software: you can redistribute it and/or modify it
// under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 2.1 of the License, or (at your
// option) any later version.  This program is distributed in the hope that it
// will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty
// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser
// General Public License for more details.  You should have received a copy
// of the GNU Lesser General Public License along with this program.  If not,
// see <http://www.gnu.org/licenses/>.
package net.opentsdb.core.tsuid;

import net.opentsdb.core.IncomingDataPoints;
import net.opentsdb.core.TSDB;
import net.opentsdb.core.TsuidExchange;
import net.opentsdb.plugin.impl.LocalMetaCache;
import net.opentsdb.uid.UniqueId;
import net.opentsdb.utils.Config;
import net.opentsdb.utils.expiringmap.ExpiringMap;
import net.opentsdb.utils.hbase.HbaseMetaUtils;
import org.hbase.async.HBaseClient;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.powermock.api.mockito.PowerMockito;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;
import org.powermock.reflect.Whitebox;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;

import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.*;

/**
 * tsuid 删除操作测试
 */
@RunWith(PowerMockRunner.class)
@PrepareForTest({TSDB.class,Config.class,HBaseClient.class,UniqueId.class,IncomingDataPoints.class})
public class TestTsuidDelete {

    HBaseClient client = mock(HBaseClient.class);
    UniqueId metrics = mock(UniqueId.class);
    UniqueId tag_names = mock(UniqueId.class);
    UniqueId tag_values = mock(UniqueId.class);
    HbaseMetaUtils metaUtils = mock(HbaseMetaUtils.class);
    LocalMetaCache metaCache = mock(LocalMetaCache.class);


    private TSDB tsdb;
    private Map<String, String> tags;
    private Config config;

    @Before
    public void before() throws Exception{
        config = new Config(false);

        config.overrideConfig("tsd.http.service","");
        config.overrideConfig("tsd.http.service.path","");
        config.overrideConfig("tsd.worker.id","");
        config.overrideConfig("tsd.core.meta.cache.enable","true");
        config.overrideConfig("tsd.timeseriesfilter.enable","true");
        config.overrideConfig("tsd.core.meta.enable_tsuid_incrementing","true");
        config.setAutoMetric(true);

        tags = new HashMap<>();
        tags.put("tagk1","tagv1");

        tsdb = new TSDB(client, config);

        Whitebox.setInternalState(tsdb.getHbaseWrapper(),"metaUtils",metaUtils);
        Whitebox.setInternalState(tsdb,"metrics",metrics);
        Whitebox.setInternalState(tsdb,"tag_names",tag_names);
        Whitebox.setInternalState(tsdb,"tag_values",tag_values);

        PowerMockito.mockStatic(IncomingDataPoints.class);

        when(metaUtils.scanTableData(any())).thenReturn(new ArrayList<>());
        when(metaUtils.getTsuidCount()).thenReturn(0L);

        when(metrics.getOrCreateId(any())).thenReturn(new byte[]{0,0,1});
        when(tag_names.getOrCreateId(any())).thenReturn(new byte[]{0,0,1});
        when(tag_values.getOrCreateId(any())).thenReturn(new byte[]{0,0,1});

        // 返回13位，还有时间戳占4位
        when(IncomingDataPoints.rowKeyTemplate(any(), any(), any())).thenReturn(new byte[]{0,0,1,0,0,0,0,0,0,1,0,0,1});

        tsdb.initializePlugins(true);
    }

    /**
     * 验证触发删除
     * {@link net.opentsdb.plugin.impl.TsuidDeleteImpl#deleteTsuid(java.lang.String)}
     */
    @Test
    public void testDeleteTsuid(){
        String tsuid = "001001001";
        Whitebox.setInternalState(tsdb.getPluginWrapper(),"meta_cache",metaCache);

        when(metaUtils.existsDelTsuid(any())).thenReturn(false);
        when(metaUtils.existsTsuid(any())).thenReturn(true);

        tsdb.deleteHbaseTSMeta(tsuid);
        tsdb.deleteMapTsuid(tsuid, true);

        verify(metaCache, times(1)).removeIncrementTsuid(any());
        verify(metaUtils,times(1)).deleteCountMeta(any());
        verify(metaUtils,times(1)).storeDelTSMeta(any());
        verify(metaUtils,times(1)).decrementDefaultAndGetCounter();
    }

    /**
     * 验证map
     * {@link LocalMetaCache#removeIncrementTsuid(byte[])}
     */
    @Test
    public void testRemoveIncrementTsuid(){
        String  metric = "metric1";
        byte[] tsuidByte = TsuidExchange.getTsuidFromData(tsdb, metric, tags);
        String tsuid = UniqueId.uidToString(tsuidByte);

        LocalMetaCache cache = Whitebox.getInternalState(tsdb.getPluginWrapper(),"meta_cache");
        ExpiringMap map = Whitebox.getInternalState(cache,"metaMap");

        tsdb.addPoint(metric, System.currentTimeMillis(), 0L, tags);

        assertTrue("map新增数据",map.size() == 1);
        assertTrue("map存在数据", map.containsKey(ByteBuffer.wrap(tsuidByte)));

        tsdb.deleteHbaseTSMeta(tsuid);
        tsdb.deleteMapTsuid(tsuid, false);

        assertTrue("map删除数据",map.size() == 0);
        assertTrue("map不存在数据", !map.containsKey(ByteBuffer.wrap(tsuidByte)));
    }

    /**
     * 验证定时器删除
     */
    @Test
    public void testTtlRemove() throws Exception{
        String  metric = "metric1";
        byte[] tsuidByte = TsuidExchange.getTsuidFromData(tsdb, metric, tags);

        LocalMetaCache cache = Whitebox.getInternalState(tsdb.getPluginWrapper(),"meta_cache");
        // ttl设为3s
        config.overrideConfig("tsd.core.meta.tsuid.ttl","3");
        // 重新加载metacache
        cache.initialize(tsdb);

        when(metaUtils.existsDelTsuid(any())).thenReturn(false);
        when(metaUtils.existsTsuid(any())).thenReturn(true);
        ExpiringMap map = Whitebox.getInternalState(cache,"metaMap");
        tsdb.addPoint(metric, System.currentTimeMillis(), 0L, tags);

        assertTrue("map新增数据",map.size() == 1);
        assertTrue("map存在数据", map.containsKey(ByteBuffer.wrap(tsuidByte)));

        Thread.sleep(3*1000);
        assertTrue("map删除数据",map.size() == 0);
        assertTrue("map不存在数据", !map.containsKey(ByteBuffer.wrap(tsuidByte)));

        verify(metaUtils,times(1)).deleteCountMeta(any());
//        verify(metaUtils,times(1)).storeDelTSMeta(any());
//        verify(metaUtils,times(1)).decrementDefaultAndGetCounter();
    }

    /**
     * 验证重设ttl
     * @throws Exception
     */
    @Test
    public void testResetRemove() throws Exception{
        String  metric = "metric1";
        byte[] tsuidByte = TsuidExchange.getTsuidFromData(tsdb, metric, tags);

        LocalMetaCache cache = Whitebox.getInternalState(tsdb.getPluginWrapper(),"meta_cache");

        when(metaUtils.existsDelTsuid(any())).thenReturn(false);
        when(metaUtils.existsTsuid(any())).thenReturn(true);
        ExpiringMap map = Whitebox.getInternalState(cache,"metaMap");
        tsdb.addPoint(metric, System.currentTimeMillis(), 0L, tags);

        assertTrue("map新增数据",map.size() == 1);
        assertTrue("map存在数据", map.containsKey(ByteBuffer.wrap(tsuidByte)));

        // 重设ttl
        cache.resetTTL(3);

        Thread.sleep(3*1000+100);
        assertTrue("map删除数据",map.size() == 0);
        assertTrue("map不存在数据", !map.containsKey(ByteBuffer.wrap(tsuidByte)));

        verify(metaUtils,times(1)).deleteCountMeta(any());
//        verify(metaUtils,times(1)).storeDelTSMeta(any());
//        verify(metaUtils,times(1)).decrementDefaultAndGetCounter();
    }

}
